package com.huawei

import org.apache.spark.rdd.RDD
import org.apache.spark.sql.{Row, SQLContext}
import org.apache.spark.sql.types.{StringType, StructField, StructType}
import org.apache.spark.storage.StorageLevel
import org.apache.spark.{SparkConf, SparkContext}

object DataStatistics {
    private val schemaString = "id,gender,height"

    def main(args: Array[String]): Unit = {
        if (args.length < 1) {
            println("Usage: DataStatistics <file>")
            System.exit(1)
        }
        val conf = new SparkConf().setAppName("DataStatistics")
        val sc = new SparkContext(conf)
        sc.setLogLevel("WARN")
        val peopleDataRdd = sc.textFile(args(0).trim)
        val sqlCtx = new SQLContext(sc)

        val schemaArr = schemaString.split(",")
        val schema = StructType(schemaArr.map(filedName => StructField(filedName, StringType, true)))
        val rowRdd: RDD[Row] = peopleDataRdd
            .map(_.split(","))
            .map(eachRow => Row(eachRow(0), eachRow(1), eachRow(2).trim))
        val peopleDF = sqlCtx.createDataFrame(rowRdd, schema)

        peopleDF.persist(StorageLevel.MEMORY_AND_DISK)
        peopleDF.createOrReplaceTempView("people")

        //  获取男性身高超过180cm的数据
        val higherMale180 = sqlCtx.sql("select * from people where height > 180 and gender = 'M'")
        println("Men whose height are higher than 180cm: " + higherMale180.count())

        // 获取男性身高超过210cm的数据，只显示前10条
        println("Men whose height is more than 210cm: ")
        peopleDF.filter(peopleDF("gender").equalTo("M")).filter(peopleDF("height").gt(210)).show(10)

        // 将身高由高到低排序，获取前10人的数据
        println("Top 10 people whose height is higher: ")
        peopleDF.sort(peopleDF("height").desc).take(10).foreach(println)

        // 获取男性平均身高
        println("The Average height for Men")
        peopleDF.filter(peopleDF("gender").equalTo("M")).agg("height" -> "avg").show()

        // 获取女性最高身高
        println("The Max height for Women: ")
        peopleDF.filter(peopleDF("gender").equalTo("F")).agg("height" -> "max").show()

        println("All the statistics are finished on structured People Data.")

    }
}
